package com.base.cn.platform.os.common;

import com.base.cn.platform.os.common.redisson.ParamEntity;
import com.base.cn.platform.os.common.redisson.RedissonConstants;
import com.base.cn.platform.os.common.utils.ObjectUtils;
import com.base.cn.platform.os.entity.user.integral.CusUserIntegral;
import com.base.cn.platform.os.service.user.integral.CusUserIntegralBiz;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.math.BigDecimal;
import java.util.Map;
import java.util.concurrent.*;

/**
 * @author : guo shi qi
 * @Date: 2019-10-23 15:49
 * @Description:
 */
@Slf4j
@Component
public class RedissonQueue {

    private Gson gson = new GsonBuilder().setDateFormat("yyyy-MM-dd HH:mm:ss").create();

    @Autowired
    private RedissonClient redissonClient;

    private RBlockingQueue<ParamEntity> blockingFairQueue;

    private RDelayedQueue<ParamEntity> delayedQueue;

    private ExecutorService singleThreadPool;
    @Autowired
    private CusUserIntegralBiz cusUserIntegralBiz;

    /**
     * 初始化 延时队列 & 线程池（程序启动初始化）
     */
    @PostConstruct
    public void initPool(){
        blockingFairQueue = redissonClient.getBlockingQueue(RedissonConstants.DELAY_QUEUE);
        delayedQueue  = redissonClient.getDelayedQueue(blockingFairQueue);
        ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("demo-pool-%d").build();
        singleThreadPool = new ThreadPoolExecutor(1, 1,0L, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<Runnable>(1024), namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());
    }
    /**
     * 添加延时队列
     * @param paramEntity 参数对象
     * @param minute 时间（分钟）
     * @param destroy 是否销毁当前对象
     */
    public void putQueue(ParamEntity paramEntity,Long minute,Boolean destroy) {
        try {
            log.info("-存入队列--param---" + gson.toJson(paramEntity) + "--minute--" + minute + "--destroy--" + destroy);
            // 一分钟以后将消息发送到指定队列
            //相当于minute分钟后取消订单
            //延迟队列包含paramEntity minute分钟，然后将其传输到blockingFairQueue中
            //在1分钟后就可以在blockingFairQueue 中获取cparamEntity了
            delayedQueue.offer(paramEntity, minute, TimeUnit.MINUTES);
            //在该对象不再需要的情况下，应该主动销毁。
            // 仅在相关的Redisson对象也需要关闭的时候可以不用主动销毁。
            if (destroy) {
                delayedQueue.destroy();
            }
            //redissonClient.shutdown();
        }catch (Exception e){
            e.printStackTrace();
            log.error("----存入队列发生错误----"+gson.toJson(paramEntity));
        }
    }
    public void getQueue(){
        singleThreadPool.execute(()-> {
            while (true) {
                ParamEntity paramEntity = null;
                try {
                    paramEntity = blockingFairQueue.take();
                    if (ObjectUtils.isNotEmpty(paramEntity)) {
                        log.info("-----开始执行队列任务------"+gson.toJson(paramEntity));
                        if (RedissonConstants.INTEGRAL_QUEUE.equals(paramEntity.getType())) {
                             log.info("--实现积分业务---内容---"+gson.toJson(paramEntity));
                            dealIntegralQueue(paramEntity);
//                            if (true) {
//                                //再次放入队列中再次执行
//                                delayedQueue.offer(paramEntity, 1, TimeUnit.MINUTES);
//                                log.error("---积分操作失败-重试--" + gson.toJson(paramEntity));
//                            }
                        }
                        if (RedissonConstants.ANSWER_REFUND_QUEUE.equals(paramEntity.getType())) {
                            log.info("---实现答疑退费业务---");
                            //实现业务，根据业务返回结果，如果失败，再此放入队列，一分钟后执行
                            Boolean result = true;
                            if (!result) {
                                //再次放入队列中再次执行
                                delayedQueue.offer(paramEntity, 1, TimeUnit.MINUTES);
                                log.error("---退费操作失败-重试--" + gson.toJson(paramEntity));
                            }
                        }
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                    log.error("---队列出现问题--" + gson.toJson(paramEntity));
                }
            }
        });
    }


    /**
     * 处理积分队列业务
     * @param paramEntity
     */
    private void dealIntegralQueue(ParamEntity paramEntity) {
        Map<String, Object> map = paramEntity.getMap();
        Integer protectScore=(Integer)map.get("protectScore");
        BigDecimal userId = new BigDecimal(map.get("userId").toString());
        CusUserIntegral cusUserIntegral = cusUserIntegralBiz.findOne(" userId= " + userId, null);
        cusUserIntegral.setProtectScore(cusUserIntegral.getProtectScore() - protectScore);
        cusUserIntegral.setCurrentScore(cusUserIntegral.getCurrentScore() + protectScore);
        cusUserIntegralBiz.updateById(cusUserIntegral, null);
    }
}

